

package io.openmessaging.demo;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import com.markhsiu.store.StoreDB;

import io.openmessaging.KeyValue;
import io.openmessaging.Message;
import io.openmessaging.MessageHeader;
import io.openmessaging.PullConsumer;

public class DefaultPullConsumer implements PullConsumer {

	private KeyValue properties;
	private String queue;
	private List<String> bucketList = new ArrayList<>();

	private int bucketListSize;
	private int bucketListIndex = 0;
	private String bucket;
	
	private boolean isReady = false;
	
	public DefaultPullConsumer(KeyValue properties) {
		this.properties = properties;
		StoreDB.builder(properties, false);
	}

	@Override
	public KeyValue properties() {
		return properties;
	}

	/**
	 * buidler
	 */
	@Override
	public Message poll() {
		// if (bucketsSize == 0 || queue == null) {
		// return null;
		// }
		if(isReady){
			 Message message = StoreDB.pull(queue);
			 if(message == null){
				 return null;
			 }
			 message.putHeaders(MessageHeader.QUEUE, queue);
			 return message;
		}

		// use Round Robin
        Message message = StoreDB.pull(bucket);
        if (message == null) {
            if (++bucketListIndex >= bucketListSize) {
            	isReady = true;
                return  poll();
            }
            bucket = bucketList.get(bucketListIndex);
            return poll();
        }
        message.putHeaders(MessageHeader.TOPIC, bucket);
        return message;
	}

	@Override
	public Message poll(KeyValue properties) {
		throw new UnsupportedOperationException("Unsupported");
	}

	@Override
	public void ack(String messageId) {
		throw new UnsupportedOperationException("Unsupported");
	}

	@Override
	public void ack(String messageId, KeyValue properties) {
		throw new UnsupportedOperationException("Unsupported");
	}

	/**
	 * buidler
	 */
	@Override
	public void attachQueue(String queueName, Collection<String> topics) {
		if (queue != null && !queue.equals(queueName)) {
			throw new RuntimeException("You have alreadly attached to a queue " + queue);
		}
		queue = queueName;
		bucketList.addAll(topics);
		bucketListSize = bucketList.size();
        bucket = bucketList.get(0);
	}

}

